package com.lagou.client;

import com.lagou.rpc.JSONSerializer;
import com.lagou.rpc.RpcEncoder;
import com.lagou.rpc.RpcRequest;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.util.internal.StringUtil;
import org.apache.curator.framework.CuratorFramework;
import org.apache.zookeeper.data.Stat;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class RpcConsumer {
    // 创建线程池对象
    private static ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private static Map<String,UserClientHandler> userClientHandlerMap = new HashMap<>();

    private static Map<String, Bootstrap> bootstrapMap = new HashMap<>();
    private static Map<String, ChannelFuture> channelFutureMap = new HashMap<>();
    private static CuratorFramework curator = Config.getCurator();

    // 创建代理对象  providerName: UserService#sayHello#are you ok?
    public Object createProxy(final Class<?> serviceClass, final String serviceName, final String methodName) {
        // 借助JDK动态代理对象
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{serviceClass}, new InvocationHandler() {
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

//                System.out.println("代理"+ method.getName());
                if(!method.getName().equals(methodName)) {
                    // 其他方法不做处理，免得发出不受控制的请求
//                    System.out.println("不是sayHello "+proxy+":"+args);
                    return method.invoke(proxy, args);
                }
                // 只代理methodName参数对应的方法
                // 调用初始化netty客户端
                if(userClientHandlerMap.isEmpty())
                    initClient();
                // 设置参数
//                userClientHandler.setPara(providerName+args[0]);
                String targetNettyServer = selectNettyServer();
                if(targetNettyServer ==null || targetNettyServer.equals("") ||  userClientHandlerMap.get(targetNettyServer)==null) {
                    System.out.println("没有可用的连接");
                }
                else {
                    System.out.println("选用连接"+targetNettyServer);
                    UserClientHandler userClientHandler = userClientHandlerMap.get(targetNettyServer);
                    userClientHandler.setPara(new RpcRequest(
                            String.valueOf(System.currentTimeMillis()),
                            serviceName,
                            methodName,
                            method.getParameterTypes(),
                            args
                    ));

                    // 去服务端请求数据
                    return executor.submit(userClientHandler).get();
                }
                return null;
            }

        });
    }
    private static String selectNettyServer() throws Exception {
        String result = "";
        long minResponseCost = Long.MAX_VALUE;
        for (String key : getNettyServers()) {
            Long responseCost = Long.parseLong(new String(curator.getData().forPath("/provider/" + key)));
            Stat stat = new Stat();
            curator.getData().storingStatIn(stat).forPath("/provider/" + key);
            if(System.currentTimeMillis() - stat.getMtime() > 5000) {
                responseCost = 0L;
            }
            if(minResponseCost > responseCost) {
                minResponseCost = responseCost;
                result = key;
            }
        }
        return result;
    }

    // 初始化Netty客户端
    public static void initClient() throws Exception {
        for(String nettyServer : getNettyServers()) {
            connect(nettyServer);
        }
        watch();

    }
    private static void connect(String nettyServer) throws Exception {
        UserClientHandler userClientHandler = new UserClientHandler();
        String host = nettyServer.split(":")[0];
        int port = Integer.parseInt(nettyServer.split(":")[1]);
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
//                        pipeline.addLast(new StringEncoder());
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new StringDecoder());
                        pipeline.addLast(userClientHandler);
                    }
                });
        ChannelFuture channelFuture = bootstrap.connect(host, port);


        channelFuture.sync();
        bootstrapMap.put(nettyServer, bootstrap);
        channelFutureMap.put(nettyServer, channelFuture);
        userClientHandlerMap.put(nettyServer, userClientHandler);
        System.out.println("建立对 "+host+":"+port+" netty服务的连接");
    }

    public static void refresh() throws Exception {
        List<String> nettyServers = getNettyServers();

        for (String existNettyServer : bootstrapMap.keySet()) {
            if(!nettyServers.contains(existNettyServer)) {
                System.out.println(existNettyServer+"下线");
                userClientHandlerMap.get(existNettyServer).close();
//                channelFutureMap.get(existNettyServer).channel().closeFuture().sync();
                String[] split = existNettyServer.split(":");
                bootstrapMap.get(existNettyServer).remoteAddress(split[0], Integer.parseInt(split[1]));
                userClientHandlerMap.put(existNettyServer,null);
                channelFutureMap.put(existNettyServer, null);
                bootstrapMap.put(existNettyServer, null);

            }
        }
        for(String nettyServer: nettyServers) {
            if(bootstrapMap.get(nettyServer)==null) {
                System.out.println(nettyServer+"上线");
                connect(nettyServer);
            }
        }
        watch();
    }

    private static List<String> getNettyServers() {

        try {
            List<String> strings = Config.getCurator().getChildren().forPath("/provider");
            return strings;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return new ArrayList<>();
    }
    private static void watch() {
        try {
            Config.getCurator().getChildren().usingWatcher(new MyCuratorWatcher()).forPath("/provider");
        }catch (Exception e) {
            e.printStackTrace();
        }
    }
}
